{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Test TCP (ttcp): The standard old-school tool for TCP performance benchmarking\n",
    "- https://en.wikipedia.org/wiki/Ttcp"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# What performance do we care?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Bandwidth, MB/s\n",
    "- Throughput, messages/s, queries/s (QPS), transactions/s (TPS)\n",
    "- Latency, milliseconds, percentiles\n",
    "- Utilization, percent, payload vs. carrier, goodput vs. theory BW\n",
    "- Overhead, eg. CPU usage, for compression and/or encryption"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "带宽; netcat; \n",
    "- 大于理论值: 是不是数据压缩; \n",
    "- 小于理论值: 消息分包等\n",
    "- 太小于理论值, 是不是有地方错误了, 特别是CPU利用率低时, 网络带宽和CPU带宽都没利用好; \n",
    "- 太小于理论值, 但CPU利用高, 100%时, 是不是计算开销比较大, 例如ssh拷贝文件时, 并行化ssh是否减小占用\n",
    "\n",
    "吞吐量:\n",
    "- 应用层面的性能指标, 每秒查询次数, 每秒发送消息, 每秒完成的事物\n",
    "\n",
    "延迟\n",
    "- 平均延迟, 百分延迟(50%, 90%)\n",
    "\n",
    "资源使用率\n",
    "- 磁盘使用率; CPU使用率; \n",
    "\n",
    "额外开销\n",
    "- 压缩, 加密"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Why do we re-implement TTCP?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- It uses all basic Sockets APIs: `socket, listen, bind, accept, connect, read/recv, write/send, shutdown, close`, etc.\n",
    "- The protocol is binary, not just byte stream, so it's better than the classic echo example\n",
    "- Typical behaviors, meaningful results, instead of packets/s\n",
    "- Service as benchmark for programming language as well, by comparing CPU usage\n",
    "- Not concurrent,at least in the very basic form"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 使用了基本的Socket API\n",
    "- 协议涉及格式, 需要分包\n",
    "- 用TCP实现的程序, 具有典型的行为; 可以针对类似TCP程序有一定学习反馈\n",
    "- 可以用多种语言的实现\n",
    "- 没有并发, 1个client, 1个server"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# The Protocol"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![2.1](./images/2.1.jpg)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 请求响应协议\n",
    "- 粗线表示数据是有效载荷的\n",
    "- Ack服务端的响应\n",
    "- 客户端发送最后一个有效载荷, 收到服务端的响应后关闭连接\n",
    "- 客户端收到Ack, 再发Payload; 比netcat只发不响应会慢一些"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// https://github.com/chenshuo/muduo/blob/master/examples/ace/ttcp/common.h\n",
    "\n",
    "struct SessionMessage\n",
    "{\n",
    "  int32_t number; // 随后要发送多少个 PayloadMessage\n",
    "  int32_t length; // 每个 PayloadMessage 内数据区 data[] 的字节数\n",
    "} __attribute__ ((__packed__));\n",
    "\n",
    "struct PayloadMessage\n",
    "{\n",
    "  int32_t length;\n",
    "  ichar data[0]; // C语言的技术; 数组长度是运行时决定的\n",
    "};"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# The Code"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Straight forward with blocking IO\n",
    "  - [muduo/examples/ace/ttcp/ttcp_blocking.cc](https://github.com/chenshuo/muduo/blob/master/examples/ace/ttcp/ttcp_blocking.cc)(C with sockets API)\n",
    "  - [recipes/tpc/bin/ttcp.cc](https://github.com/chenshuo/recipes/blob/master/tpc/bin/ttcp.cc)(C++ with a thin wrapper)\n",
    "  - [muduo-examples-in-go/examples/ace/ttcp/ttcp.go](https://github.com/chenshuo/muduo-examples-in-go/blob/master/examples/ace/ttcp/ttcp.go) (Go)\n",
    "- Non-blocking IO with muduo library\n",
    "  - [muduo/examples/ace/ttcp/ttcp.cc](https://github.com/chenshuo/muduo/blob/master/examples/ace/ttcp/ttcp.cc): (稍微复杂, 并且没有提高性能)\n",
    "- None of above support concurrentconnections\n",
    "  - Pretty easy to enable, thread-per-connection for first three."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## [ttcp_blocking.cc](https://github.com/chenshuo/muduo/blob/master/examples/ace/ttcp/ttcp_blocking.cc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "void transmit(const Options& opt);\n",
    "void receive(const Options& opt);\n",
    "\n",
    "static int write_n(int sockfd, const void* buf, int length);\n",
    "static int read_n(int sockfd, void* buf, int length);\n",
    "\n",
    "static int acceptOrDie(uint16_t port);"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 服务端\n",
    "void receive(const Options& opt)\n",
    "{\n",
    "  // 接受一个连接\n",
    "  int sockfd = acceptOrDie(opt.port); \n",
    "\n",
    "  struct SessionMessage sessionMessage = { 0, 0 };\n",
    "  // 客户端发过来一个SessionMessage, 8个字节的消息\n",
    "  if (read_n(sockfd, &sessionMessage, sizeof(sessionMessage)) != sizeof(sessionMessage))\n",
    "  {\n",
    "    perror(\"read SessionMessage\");\n",
    "    exit(1);\n",
    "  }\n",
    "\n",
    "  // 消息是网络字节序, 转为本机字节序 \n",
    "  sessionMessage.number = ntohl(sessionMessage.number);\n",
    "  sessionMessage.length = ntohl(sessionMessage.length);\n",
    "  // 服务端打印\n",
    "  printf(\"receive number = %d\\nreceive length = %d\\n\",\n",
    "         sessionMessage.number, sessionMessage.length);\n",
    "  // 总长度; \n",
    "  const int total_len = static_cast<int>(sizeof(int32_t) + sessionMessage.length);\n",
    "  \n",
    "  // 有安全漏洞length很大 ,造成malloc分配大内存, 造成拒绝响应攻击; TTCP是内网工具; 可以限制length的长度\n",
    "  PayloadMessage* payload = static_cast<PayloadMessage*>(::malloc(total_len)); \n",
    "  assert(payload);\n",
    "\n",
    "  // 循环, 服务端代码的剩余部分; 循环n(消息数)次\n",
    "  for (int i = 0; i < sessionMessage.number; ++i)\n",
    "  {\n",
    "    payload->length = 0;\n",
    "    // 读取4个字节\n",
    "    if (read_n(sockfd, &payload->length, sizeof(payload->length)) != sizeof(payload->length))\n",
    "    {\n",
    "      perror(\"read length\");\n",
    "      exit(1);\n",
    "    }\n",
    "    // 转为本机序\n",
    "    payload->length = ntohl(payload->length);\n",
    "    // 验证长度相等\n",
    "    assert(payload->length == sessionMessage.length);\n",
    "    // 读取payload->length个字节\n",
    "    if (read_n(sockfd, payload->data, payload->length) != payload->length)\n",
    "    {\n",
    "      perror(\"read payload data\");\n",
    "      exit(1);\n",
    "    }\n",
    "    // 转为本机序\n",
    "    int32_t ack = htonl(payload->length);\n",
    "    // 构造响应\n",
    "    if (write_n(sockfd, &ack, sizeof(ack)) != sizeof(ack))\n",
    "    {\n",
    "      perror(\"write ack\");\n",
    "      exit(1);\n",
    "    }\n",
    "  }\n",
    "  // 释放内存\n",
    "  ::free(payload);\n",
    "  // 关闭Socket连接\n",
    "  ::close(sockfd);\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> \"验证长度相等\": 是对协议正确性和内存安全的最后一道防线。一旦收到的 `PayloadMessage` 长度与事先协商的不一致，程序立刻中止，提醒开发者或测试人员排查网络传输错误、协议实现漏洞或潜在的恶意输入"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 客户端\n",
    "void transmit(const Options& opt)\n",
    "{\n",
    "  struct sockaddr_in addr = resolveOrDie(opt.host.c_str(), opt.port);\n",
    "  printf(\"connecting to %s:%d\\n\", inet_ntoa(addr.sin_addr), opt.port);\n",
    "\n",
    "  int sockfd = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);\n",
    "  assert(sockfd >= 0);\n",
    "  int ret = ::connect(sockfd, reinterpret_cast<struct sockaddr*>(&addr), sizeof(addr));\n",
    "  if (ret)\n",
    "  {\n",
    "    perror(\"connect\");\n",
    "    printf(\"Unable to connect %s\\n\", opt.host.c_str());\n",
    "    ::close(sockfd);\n",
    "    return;\n",
    "  }\n",
    "\n",
    "  printf(\"connected\\n\");\n",
    "  muduo::Timestamp start(muduo::Timestamp::now());\n",
    "  struct SessionMessage sessionMessage = { 0, 0 };\n",
    "  sessionMessage.number = htonl(opt.number);\n",
    "  sessionMessage.length = htonl(opt.length);\n",
    "  if (write_n(sockfd, &sessionMessage, sizeof(sessionMessage)) != sizeof(sessionMessage))\n",
    "  {\n",
    "    perror(\"write SessionMessage\");\n",
    "    exit(1);\n",
    "  }\n",
    "\n",
    "  const int total_len = static_cast<int>(sizeof(int32_t) + opt.length);\n",
    "  PayloadMessage* payload = static_cast<PayloadMessage*>(::malloc(total_len));\n",
    "  assert(payload);\n",
    "  payload->length = htonl(opt.length);\n",
    "  for (int i = 0; i < opt.length; ++i)\n",
    "  {\n",
    "    payload->data[i] = \"0123456789ABCDEF\"[i % 16];\n",
    "  }\n",
    "\n",
    "  double total_mb = 1.0 * opt.length * opt.number / 1024 / 1024;\n",
    "  printf(\"%.3f MiB in total\\n\", total_mb);\n",
    "\n",
    "  for (int i = 0; i < opt.number; ++i)\n",
    "  {\n",
    "    int nw = write_n(sockfd, payload, total_len);\n",
    "    assert(nw == total_len);\n",
    "\n",
    "    int ack = 0;\n",
    "    int nr = read_n(sockfd, &ack, sizeof(ack));\n",
    "    assert(nr == sizeof(ack));\n",
    "    ack = ntohl(ack);\n",
    "    assert(ack == opt.length);\n",
    "  }\n",
    "\n",
    "  ::free(payload);\n",
    "  ::close(sockfd);\n",
    "  double elapsed = timeDifference(muduo::Timestamp::now(), start);\n",
    "  printf(\"%.3f seconds\\n%.3f MiB/s\\n\", elapsed, total_mb / elapsed);\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## [ttcp.cc](https://github.com/chenshuo/recipes/blob/master/tpc/bin/ttcp.cc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 客户端\n",
    "\n",
    "void transmit(const Options& opt)\n",
    "{\n",
    "  InetAddress addr;\n",
    "  // 获取地址端口; 解析主机名为IP地址\n",
    "  if (!InetAddress::resolve(opt.host.c_str(), opt.port, &addr))\n",
    "  {\n",
    "    printf(\"Unable to resolve %s\\n\", opt.host.c_str());\n",
    "    return;\n",
    "  }\n",
    "\n",
    "  printf(\"connecting to %s\\n\", addr.toIpPort().c_str());\n",
    "  // 连接, 成功后会返回一个string(使用了移动语义)\n",
    "  TcpStreamPtr stream(TcpStream::connect(addr)); //RAII\n",
    "  if (!stream)\n",
    "  {\n",
    "    printf(\"Unable to connect %s\\n\", addr.toIpPort().c_str());\n",
    "    perror(\"\");\n",
    "    return;\n",
    "  }\n",
    "\n",
    "  // 禁用TCP_NODELAY: 禁用 Nagle 算法后，TCP 将不再等待合并小包，也不再等待 TCP ACK，而是立即发送数据\n",
    "  // 发送消息不用等待ACK: 指TCP协议本身的\"ACK\", 不是应用层消息的\"ACK\"\n",
    "  if (opt.nodelay)\n",
    "  {\n",
    "    stream->setTcpNoDelay(true);\n",
    "  }\n",
    "  printf(\"connected\\n\");\n",
    "  double start = now();\n",
    "  // 构造SessionMessage\n",
    "  struct SessionMessage sessionMessage = { 0, 0 };\n",
    "  sessionMessage.number = htonl(opt.number);\n",
    "  sessionMessage.length = htonl(opt.length);\n",
    "  // 发送整个消息; 返回的是发送的字节数, 如果与预期的字节数不等, 会报错\n",
    "  if (stream->sendAll(&sessionMessage, sizeof(sessionMessage)) != sizeof(sessionMessage))\n",
    "  {\n",
    "    perror(\"write SessionMessage\");\n",
    "    return;\n",
    "  }\n",
    "\n",
    "  // 构造PayloadMessage\n",
    "  const int total_len = sizeof(int32_t) + opt.length;\n",
    "  PayloadMessage* payload = static_cast<PayloadMessage*>(::malloc(total_len));\n",
    "  std::unique_ptr<PayloadMessage, void (*)(void*)> freeIt(payload, ::free); // C++11, 自动调用free\n",
    "  assert(payload);\n",
    "  payload->length = htonl(opt.length);\n",
    "  // 构造数据\n",
    "  for (int i = 0; i < opt.length; ++i)\n",
    "  {\n",
    "    payload->data[i] = \"0123456789ABCDEF\"[i % 16]; // [i % 16]数字对应单字符\n",
    "  }\n",
    "\n",
    "  // 总共发送多少兆的数据\n",
    "  double total_mb = 1.0 * opt.length * opt.number / 1024 / 1024;\n",
    "  printf(\"%.3f MiB in total\\n\", total_mb);\n",
    "\n",
    "  for (int i = 0; i < opt.number; ++i)\n",
    "  {\n",
    "    // 法\n",
    "    int nw = stream->sendAll(payload, total_len);\n",
    "    assert(nw == total_len);\n",
    "\n",
    "    int ack = 0;\n",
    "    // 等待应用层服务端的ACK\n",
    "    int nr = stream->receiveAll(&ack, sizeof(ack));\n",
    "    assert(nr == sizeof(ack));\n",
    "    ack = ntohl(ack);\n",
    "    assert(ack == opt.length);\n",
    "  }\n",
    "\n",
    "  double elapsed = now() - start;\n",
    "  printf(\"%.3f seconds\\n%.3f MiB/s\\n\", elapsed, total_mb / elapsed);\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> Nagle算法会将小数据包聚合后再发送，减少网络中的小包数量，从而提高传输效率，但会增加延迟"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Test environment"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A: atom is Intel Celeron 1037U\n",
    "- Low end Ivy Bridge dual core @ 1.8 GHz,8GB ram\n",
    "\n",
    "B: e6400 is Dell laptop 2009\n",
    "- Intel Core2 Duo P8600 @2.4 GHz, 4GB ram\n",
    "\n",
    "C: e350 is AMD E-350\n",
    "- 8GB ram, lowend APU"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Multidimensional performance results"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Host(s) to run\n",
    "- Client/server programs to run\n",
    "- Message size in byte\n",
    "- Number of messages\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| client/Server | A    | B    | C    |\n",
    "| ------------- | ---- | ---- | ---- |\n",
    "| A             |      |      |      |\n",
    "| B             |      |      |      |\n",
    "| C             |      |      |      |\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| Client/Server | C    | C++  | Go   | muduo |\n",
    "| ------------- | ---- | ---- | ---- | ----- |\n",
    "| C             |      |      |      |       |\n",
    "| C++           |      |      |      |       |\n",
    "| Go            |      |      |      |       |\n",
    "| muduo         |      |      |      |       |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 还可以加个维度变为三维表格, 第三维度是机器数(hosts)\n",
    "- 再加1个维度ttcp运行有参数(消息个数n, 消息长度l): 可以让n*l为1个G, 来变化n, l\n",
    "- 再加1个维度延迟参数: ttcp跟延迟有关; netcat只跟带宽有关"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 运行测试程序"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### client(e6400)<->server(atom)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "server: \n",
    "- `./ttcp -t e6400`\n",
    "\n",
    "client: \n",
    "- `while true; do ./ttcp -r; done`(不用每次重启客户端)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "变更消息长度\n",
    "- `./ttcp -t e6400 -l 1024`\n",
    "- `./ttcp -t e6400 -l 2048`\n",
    "- `./ttcp -t e6400 -l 4096`\n",
    "- `./ttcp -t e6400 -l 8192`\n",
    "- `./ttcp -t e6400 -l 18384`\n",
    "- `./ttcp -t e6400 -l 32768`\n",
    "- `./ttcp -t e6400 -l 65536`\n",
    "- `./ttcp -t e6400 -l ...`\n",
    "\n",
    "> 运行时间太短, 可能会受到TCP slow start的影响"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "变更消息数量\n",
    "- `./ttcp -t e6400 -l 512000 -n 4096`\n",
    "\n",
    "> 消息发送的越小, ttcp传输的延迟影响越大"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### client(atom)<->server(atom)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> 进程间通讯TCP性能也很强"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Blocking IO could block, forever"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> 阻塞IO, 可能会阻塞, 而且可能会一直阻塞下去"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## server: [echo.cc](https://github.com/chenshuo/recipes/blob/master/tpc/bin/echo.cc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "#include \"Acceptor.h\"\n",
    "#include \"InetAddress.h\"\n",
    "#include \"TcpStream.h\"\n",
    "\n",
    "#include <thread>\n",
    "#include <string.h>\n",
    "\n",
    "// a thread-per-connection current echo server\n",
    "int main(int argc, char* argv[])\n",
    "{\n",
    "  bool nodelay = false;\n",
    "  bool ipv6 = false;\n",
    "  for (int i = 1; i < argc; ++i) {\n",
    "    if (strcmp(argv[i], \"-6\") == 0)\n",
    "      ipv6 = true;\n",
    "    if (strcmp(argv[i], \"-D\") == 0)\n",
    "      nodelay = true;\n",
    "  }\n",
    "  InetAddress listenAddr(3007, ipv6);\n",
    "  Acceptor acceptor(listenAddr);\n",
    "  printf(\"Listen on port 3007\\n\");\n",
    "  printf(\"Accepting... Ctrl-C to exit\\n\");\n",
    "  int count = 0;\n",
    "\n",
    "  while (true)\n",
    "  {\n",
    "    TcpStreamPtr tcpStream = acceptor.accept();\n",
    "    printf(\"accepted no. %d client\\n\", ++count);\n",
    "    if (nodelay)\n",
    "      tcpStream->setTcpNoDelay(true);\n",
    "\n",
    "    // C++11 doesn't allow capturing unique_ptr in lambda, C++14 allows.\n",
    "    // 启动线程\n",
    "    std::thread thr([count] (TcpStreamPtr stream) {\n",
    "      printf(\"thread for no. %d client started.\\n\", count);\n",
    "      char buf[4096];\n",
    "      int nr = 0;\n",
    "      // 读取数据\n",
    "      while ( (nr = stream->receiveSome(buf, sizeof(buf))) > 0)\n",
    "      {\n",
    "        // 回显\n",
    "        int nw = stream->sendAll(buf, nr);\n",
    "        if (nw < nr)\n",
    "        {\n",
    "          break;\n",
    "        }\n",
    "      }\n",
    "      printf(\"thread for no. %d client ended.\\n\", count);\n",
    "    }, std::move(tcpStream));\n",
    "    thr.detach();\n",
    "  }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## client: [echo_client.cc](https://github.com/chenshuo/recipes/blob/master/tpc/bin/echo_client.cc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "#include \"InetAddress.h\"\n",
    "#include \"TcpStream.h\"\n",
    "#include <unistd.h>\n",
    "\n",
    "int main(int argc, const char* argv[])\n",
    "{\n",
    "  if (argc < 3)\n",
    "  {\n",
    "    printf(\"Usage: %s hostname message_length [scp]\\n\", argv[0]);\n",
    "    return 0;\n",
    "  }\n",
    "\n",
    "  const int len = atoi(argv[2]);\n",
    "\n",
    "  uint16_t port = 3007;\n",
    "  InetAddress addr;\n",
    "  if (!InetAddress::resolve(argv[1], port, &addr))\n",
    "  {\n",
    "    printf(\"Unable to resolve %s\\n\", argv[1]);\n",
    "    return 0;\n",
    "  }\n",
    "\n",
    "  printf(\"connecting to %s\\n\", addr.toIpPort().c_str());\n",
    "  TcpStreamPtr stream(TcpStream::connect(addr));\n",
    "  if (!stream)\n",
    "  {\n",
    "    perror(\"\");\n",
    "    printf(\"Unable to connect %s\\n\", addr.toIpPort().c_str());\n",
    "    return 0;\n",
    "  }\n",
    "\n",
    "  printf(\"connected, sending %d bytes\\n\", len);\n",
    "\n",
    "  std::string message(len, 'S'); // 由命令行制定\n",
    "  int nw = stream->sendAll(message.c_str(), message.size());\n",
    "  printf(\"sent %d bytes\\n\", nw);\n",
    "\n",
    "  if (argc > 3)\n",
    "  {\n",
    "    for (char cmd : std::string(argv[3]))\n",
    "    {\n",
    "      if (cmd == 's')  // shutdown\n",
    "      {\n",
    "        printf(\"shutdown write\\n\");\n",
    "        stream->shutdownWrite();\n",
    "      }\n",
    "      else if (cmd == 'p') // pause\n",
    "      {\n",
    "        printf(\"sleeping for 10 seconds\\n\");\n",
    "        ::sleep(10);\n",
    "        printf(\"done\\n\");\n",
    "      }\n",
    "      else if (cmd == 'c') // close\n",
    "      {\n",
    "        printf(\"close without reading response\\n\");\n",
    "        return 0;\n",
    "      }\n",
    "      else\n",
    "      {\n",
    "        printf(\"unknown command '%c'\\n\", cmd);\n",
    "      }\n",
    "    }\n",
    "  }\n",
    "\n",
    "  // 准备相同长度的缓冲区\n",
    "  std::vector<char> receive(len);\n",
    "  // 读取服务端消息\n",
    "  int nr = stream->receiveAll(receive.data(), receive.size());\n",
    "  printf(\"received %d bytes\\n\", nr);\n",
    "  if (nr != nw)\n",
    "  {\n",
    "    printf(\"!!! Incomplete response !!!\\n\");\n",
    "  }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "这样的程序会阻塞吗?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 测试"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```shell\n",
    "./echo\n",
    "```\n",
    "\n",
    "- 变化消息大小\n",
    "```shell\n",
    "./echo_client localhost 1024000\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "为什么会阻塞?\n",
    "\n",
    "使用`netstat -tpn` | grep `3007\\|^[AP]`查看"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![2.3](./images/2.2.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "注意查看收发队列的长度: \n",
    "- 服务器接收队列有6MB字节的数据, 发送队列有2.6MB的数据在内核缓冲区\n",
    "- 客户端接收队列有900K的数据, 发送队列有4MB的数据"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![2.3](./images/2.3.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`C Send -> S Recv`: \n",
    "- 客户端发送, 数据进入内核缓冲区, 缓冲区有大小限制, 缓冲区满了会阻塞;\n",
    "\n",
    "`C Recv <- S Send`: \n",
    "- 服务端读没有问题, 但是服务端会向客户端发送数据; \n",
    "- 当客户端向服务端发送20M, 没有发送完毕之前, 客户端是不会读取数据的; 假设缓冲区只有10M, 服务端发送完10M, 服务端阻塞在send上, 等待客户端读取, 但是客户端没有send完毕数据, 至此服务端也不会recv了"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "阻塞IO, 死锁是无法自我解脱的"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "TCP内存缓冲区大小跟内核参数有关\n",
    "- `sysctl -A | grep tcp.mem`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "阻塞这个问题应该客户端负责, 还是服务端负责\n",
    "- 客户端发送20MB的消息是正常的\n",
    "- 服务器: echo读取4k, 然后发送给客户端, 问题在于服务器没有完整读取到客户端发来的请求; \n",
    "- 正常设计协议的时候, 客户端发来header, 告诉服务器端请求的大小, 然后发送请求本身, 服务器收到header后, 准备缓冲区接收请求, 计算响应, 然后发给客户端, 客户端会收到header响应的大小为止\n",
    "- 字节流协议, 需要加入分包的措施"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# TCP self-connection"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- net.ipv4.ip_local_port_range\n",
    "- [recipes/python/self-connect.py](https://github.com/chenshuo/recipes/blob/master/python/self-connect.py)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    ">  Linux TCP协议实现中的特性，曾被误报为bug但被内核维护人员确认是正常行为"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "python"
    }
   },
   "outputs": [],
   "source": [
    "#!/usr/bin/python\n",
    "\n",
    "import errno\n",
    "import socket\n",
    "import sys\n",
    "import time\n",
    "\n",
    "if len(sys.argv) < 2:\n",
    "    print \"Usage: %s port\" % sys.argv[0]\n",
    "    print \"port should in net.ipv4.ip_local_port_range\"\n",
    "else:\n",
    "    port = int(sys.argv[1])\n",
    "    for i in range(65536):\n",
    "        try:\n",
    "            sock = socket.create_connection(('localhost', port)) # 命令行输入的端口号, 对这个端口号发起连接\n",
    "            print \"connected\", sock.getsockname(), sock.getpeername()\n",
    "            time.sleep(60*60)\n",
    "        except socket.error, e:\n",
    "            if e.errno != errno.ECONNREFUSED:\n",
    "                break"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "查看本机端口号: `netstat -ltnp`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```shell\n",
    "python self-connect.py 22\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```shell\n",
    "python self-connect.py 31000\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```shell\n",
    "sysctl -A | grep range # 查看本机端口号\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```shell\n",
    "python self-connect.py 65535\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```shell\n",
    "netstat -tpn | grep 65535\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```shell\n",
    "netstat -tpn | grep 1000\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "为什么会出现自连接\n",
    "- TCP在发起一个连接时, 会选择一个临时端口号, 从`ip_local_port_range`中选一个, 按照计数器方式选择, 选定后, 再向服务器端口发请求\n",
    "- 没有服务器进程监听端口号时, 客户端选择端口`31000`, 这时其中`127.0.0.1:31000`已经在内核中存在了, 然后向服务器发送`127.0.0.1:31000`, 内核接收到SYN, 同时有一个`127.0.0.1:31000`已经打开, 这个请求不会被拒绝, 产生RST."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "解决办法:\n",
    "- https://github.com/chenshuo/recipes/blob/master/tpc/lib/TcpStream.cc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "#include \"TcpStream.h\"\n",
    "#include \"InetAddress.h\"\n",
    "\n",
    "#include <errno.h>\n",
    "#include <signal.h>\n",
    "#include <unistd.h>\n",
    "#include <sys/socket.h>\n",
    "\n",
    "namespace\n",
    "{\n",
    "\n",
    "class IgnoreSigPipe\n",
    "{\n",
    " public:\n",
    "  IgnoreSigPipe()\n",
    "  {\n",
    "    ::signal(SIGPIPE, SIG_IGN);\n",
    "  }\n",
    "} initObj;\n",
    "\n",
    "// 判断是否是自连接\n",
    "bool isSelfConnection(const Socket& sock)\n",
    "{\n",
    "  return sock.getLocalAddr() == sock.getPeerAddr(); // 本机地址等于对方地址\n",
    "}\n",
    "\n",
    "}\n",
    "\n",
    "TcpStream::TcpStream(Socket&& sock)\n",
    "  : sock_(std::move(sock))\n",
    "{\n",
    "}\n",
    "\n",
    "int TcpStream::receiveAll(void* buf, int len)\n",
    "{\n",
    "#ifdef TEMP_FAILURE_RETRY\n",
    "  return TEMP_FAILURE_RETRY(::recv(sock_.fd(), buf, len, MSG_WAITALL));\n",
    "#else\n",
    "  return ::recv(sock_.fd(), buf, len, MSG_WAITALL);\n",
    "#endif\n",
    "}\n",
    "\n",
    "int TcpStream::receiveSome(void* buf, int len)\n",
    "{\n",
    "  return sock_.recv(buf, len);\n",
    "}\n",
    "\n",
    "int TcpStream::sendAll(const void* buf, int len)\n",
    "{\n",
    "  int written = 0;\n",
    "  while (written < len)\n",
    "  {\n",
    "    int nw = sock_.send(static_cast<const char*>(buf) + written, len - written);\n",
    "    if (nw > 0)\n",
    "    {\n",
    "      written += nw;\n",
    "    }\n",
    "    else\n",
    "    {\n",
    "      break;\n",
    "    }\n",
    "  }\n",
    "  return written;\n",
    "}\n",
    "\n",
    "int TcpStream::sendSome(const void* buf, int len)\n",
    "{\n",
    "  return sock_.send(buf, len);\n",
    "}\n",
    "\n",
    "void TcpStream::setTcpNoDelay(bool on)\n",
    "{\n",
    "  sock_.setTcpNoDelay(on);\n",
    "}\n",
    "\n",
    "void TcpStream::shutdownWrite()\n",
    "{\n",
    "  sock_.shutdownWrite();\n",
    "}\n",
    "\n",
    "TcpStreamPtr TcpStream::connect(const InetAddress& serverAddr)\n",
    "{\n",
    "  return connectInternal(serverAddr, nullptr);\n",
    "}\n",
    "\n",
    "TcpStreamPtr TcpStream::connect(const InetAddress& serverAddr, const InetAddress& localAddr)\n",
    "{\n",
    "  return connectInternal(serverAddr, &localAddr);\n",
    "}\n",
    "\n",
    "TcpStreamPtr TcpStream::connectInternal(const InetAddress& serverAddr, const InetAddress* localAddr)\n",
    "{\n",
    "  TcpStreamPtr stream;\n",
    "  Socket sock(Socket::createTCP(serverAddr.family()));\n",
    "  if (localAddr)\n",
    "  {\n",
    "    sock.bindOrDie(*localAddr);\n",
    "  }\n",
    "\n",
    "  if (sock.connect(serverAddr) == 0 && !isSelfConnection(sock))\n",
    "  {\n",
    "    // FIXME: do poll(POLLOUT) to check errors\n",
    "    stream.reset(new TcpStream(std::move(sock)));\n",
    "  }\n",
    "  return stream;\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Externsions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Measure throughput (messages/s) and latency\n",
    "- Server supports concurrent clients\n",
    "- Multiple sessions/connections, test fairness of TCP\n",
    "- Long-fat pipe,cross data centers\n",
    "  - Pipelining\n",
    "- Nagle’s algorithm and TCP slow start\n",
    "- How do we detect TCP errors?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 测试消息本身的吞吐量; 测试延迟(平均, 百分位)\n",
    "- 支持并发连接; \n",
    "- 增加客户端的并发连接; 测试TCP的公平性\n",
    "- 延迟比较大, 如何跑满带宽\n",
    "- nagle算法; tcp的慢启动\n",
    "- 如何检测TCP的错误"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Pipelining"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![2.4](./images/2.4.png)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "C++17",
   "language": "C++17",
   "name": "xcpp17"
  },
  "language_info": {
   "name": "C++17"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
